package baseFx.common.mq.rabbitmq;

import baseFx.common.Action1T;
import baseFx.common.concurrent.Task;
import baseFx.common.utils.ExceptionUtils;
import baseFx.common.utils.FxDate;
import baseFx.common.utils.RandomUtils;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;


public class RabbitMqContext {
    private ConnectionFactory factory;

    public RabbitMqContext(String host, int port, String username, String password) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(host);
        factory.setPort(port);
        factory.setUsername(username);
        factory.setPassword(password);
        factory.setShutdownTimeout(FxDate.SECONDS * 60);
        this.factory = factory;
    }

    public Connection getConnection() {
        while (true) {
            try {
                return factory.newConnection();
            } catch (IOException e) {
                throw ExceptionUtils.toRuntime(e);
            } catch (TimeoutException e) {
                Task.delay(FxDate.SECONDS * Integer.parseInt(RandomUtils.numByRange(1, 60)));
            }
        }
    }

    public long messageCount(String queue) {
        try (Connection connection = getConnection();
             Channel channel = connection.createChannel()) {
            return channel.messageCount(queue);
        } catch (TimeoutException | IOException e) {
            throw ExceptionUtils.toRuntime(e);
        }
    }

    public void useChannel(Action1T<Channel> action) {
        try (Connection connection = getConnection();
             Channel channel = connection.createChannel()) {
            action.execute(channel);
        } catch (TimeoutException | IOException e) {
            throw ExceptionUtils.toRuntime(e);
        }
    }

    public boolean existsExchange(String exchange) {
        try (Connection connection = getConnection();
             Channel channel = connection.createChannel()) {
            channel.exchangeDeclarePassive(exchange);
            return true;
        } catch (Exception e) {
            if (e.getCause() != null && e.getCause().getMessage().contains("404")) {
                return false;
            }
            throw ExceptionUtils.toRuntime(e);
        }
    }

    public boolean existsQueue(String queue) {
        try (Connection connection = getConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclarePassive(queue);
            return true;
        } catch (Exception e) {
            if (e.getCause() != null && e.getCause().getMessage().contains("404")) {
                return false;
            }
            throw ExceptionUtils.toRuntime(e);
        }
    }

    public void exchangeDeclare(String exchange, BuiltinExchangeType type) {
        exchangeDeclare(exchange, type, true);
    }

    public void exchangeDeclare(String exchange, BuiltinExchangeType type, boolean isPersist) {
        useChannel(ch -> {
            try {
                ch.exchangeDeclare(exchange, type, isPersist);
            } catch (IOException e) {
                throw ExceptionUtils.toRuntime(e);
            }
        });
    }

    public void queueDeclare(String name, boolean isPersist) {
        useChannel(ch -> {
            try {
                ch.queueDeclare(name, isPersist, false, false, null);
            } catch (IOException e) {
                throw ExceptionUtils.toRuntime(e);
            }
        });
    }

    public void queueBind(String exchange, String queue) {
        queueBind(exchange, "", queue);
    }

    public void queueBind(String exchange, String routing, String queue) {
        if (existsExchange(exchange) == false) {
            throw new RuntimeException(exchange + " 不存在");
        }
        useChannel(ch -> {
            try {
                ch.queueBind(queue, exchange, routing);
            } catch (IOException e) {
                throw ExceptionUtils.toRuntime(e);
            }
        });
    }

    public void declare(String exchange, BuiltinExchangeType type, String routing, String queue) {
        useChannel(ch -> {
            try {
                ch.exchangeDeclare(exchange, type, true);
                ch.queueDeclare(queue, true, false, false, null);
                ch.queueBind(queue, exchange, routing);
            } catch (IOException e) {
                throw ExceptionUtils.toRuntime(e);
            }
        });
    }

    public void declare(String declareString) {
        String[] items = declareString.split("&");
        if (items.length != 3 && items.length != 4) {
            throw new RuntimeException("declareString的正确格式是：exchange&BuiltinExchangeType[&routing]&queue");
        }
        String ex = items[0];
        BuiltinExchangeType type = BuiltinExchangeType.valueOf(items[1]);
        String routing = items.length == 3 ? "" : items[2];
        String queue = items.length == 3 ? items[2] : items[3];
        if (existsExchange(ex) == false) {
            declare(ex, type, routing, queue);
        }
    }
}
